package com.atguigu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.bean.TradePaymentGMVBean;
import com.atguigu.realtime.utils.ClickHouseUtil;
import com.atguigu.realtime.utils.DateFormatUtil;
import com.atguigu.realtime.utils.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.*;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.HashSet;
import java.util.Set;

public class DwsTradePaymentGMV {
    /*public static void main(String[] args) throws Exception {
        //TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //TODO 2 从Kafka dwd_trade_payment_info 主题读取数据
        String topicName = "dwd_trade_payment_info";
        String groupID = "dws_trade_payment_GMV";
        FlinkKafkaConsumer<String> kafkaConsumer = KafkaUtil.getKafkaConsumer(topicName, groupID);
        DataStreamSource<String> tradePaymentStream = env.addSource(kafkaConsumer);

        //TODO 3 转化结构为JSON
        SingleOutputStreamOperator<JSONObject> jsonObjStream = tradePaymentStream.map(JSON::parseObject);

        //TODO 4 按照sc分组
        KeyedStream<JSONObject, String> keyedStream = jsonObjStream.keyBy(jsonObj -> jsonObj.getString("sc"));
        SingleOutputStreamOperator<TradePaymentGMVBean> beanStream = keyedStream.map(new RichMapFunction<JSONObject, TradePaymentGMVBean>() {
            @Override
            public TradePaymentGMVBean map(JSONObject value) throws Exception {
                Long ts = value.getLong("ts");
                return new TradePaymentGMVBean("", "", Long.parseLong(value.getString("final_amount")), 1L, ts);
            }

            ValueState<String> idState = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> createTimeDescriptor = new ValueStateDescriptor<>("create_time", String.class);
                createTimeDescriptor.enableTimeToLive(StateTtlConfig
                        .newBuilder(Time.days(1L))
                        .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                        .build());
                idState = getRuntimeContext().getState(createTimeDescriptor);
            }
        });

        //TODO 5 开窗聚合数据
        SingleOutputStreamOperator<TradePaymentGMVBean> beanWithWatermarkStream = beanStream.assignTimestampsAndWatermarks(WatermarkStrategy.<TradePaymentGMVBean>forBoundedOutOfOrderness(Duration.ofSeconds(2L)).withTimestampAssigner(new SerializableTimestampAssigner<TradePaymentGMVBean>() {
            @Override
            public long extractTimestamp(TradePaymentGMVBean element, long recordTimestamp) {
                return element.getTs();
            }
        }));

        SingleOutputStreamOperator<TradePaymentGMVBean> reduceStream = beanWithWatermarkStream.windowAll(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10L)))
                .reduce(new ReduceFunction<TradePaymentGMVBean>() {
                    @Override
                    public TradePaymentGMVBean reduce(TradePaymentGMVBean value1, TradePaymentGMVBean value2) throws Exception {
                        value1.setFinal_amount(value1.getFinal_amount() + value2.getFinal_amount());
                        value1.setId(value1.getId() + value2.getId());
                        return value1;
                    }
                }, new AllWindowFunction<TradePaymentGMVBean, TradePaymentGMVBean, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow window, Iterable<TradePaymentGMVBean> values, Collector<TradePaymentGMVBean> out) throws Exception {
                        TradePaymentGMVBean tradePaymentGMVBean = values.iterator().next();
                        tradePaymentGMVBean.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                        tradePaymentGMVBean.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                        tradePaymentGMVBean.setTs(System.currentTimeMillis());
                        out.collect(tradePaymentGMVBean);
                    }
                });

        //TODO 6 写出到ClickHouse中
        reduceStream.addSink(ClickHouseUtil.getClickHouseSinkFunc("insert into dws_trade_payment_GMV values(?,?,?,?,?)"));

        //TODO 7 执行任务
        env.execute(groupID);
    }*/

    public static void main(String[] args) throws Exception {
        //TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //TODO 2 从Kafka dwd_trade_payment_info 主题读取数据
        String topicName = "dwd_trade_order_detail";
        String groupID = "dws_trade_payment_GMV";
        FlinkKafkaConsumer<String> kafkaConsumer = KafkaUtil.getKafkaConsumer(topicName, groupID);
        FlinkKafkaConsumer<String> KafkaConsumer2 = KafkaUtil.getKafkaConsumer("dwd_page_traffic_log", groupID);
        DataStreamSource<String> tradePaymentStream = env.addSource(kafkaConsumer);
        DataStreamSource<String> tradePaymentStream2 = env.addSource(KafkaConsumer2);


        //TODO 3 转化为对应的JavaBean
        SingleOutputStreamOperator<JSONObject> jsonObjeStream = tradePaymentStream.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                out.collect(jsonObject);
            }
        });
        SingleOutputStreamOperator<JSONObject> jsonObjWitnWatermarkStream = jsonObjeStream.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
            @Override
            public long extractTimestamp(JSONObject element, long recordTimestamp) {
                return element.getLong("ts");
            }
        }));
        KeyedStream<JSONObject, String> keyedStream = jsonObjWitnWatermarkStream.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject value) throws Exception {
                return value.getString("sc");
            }
        });
        SingleOutputStreamOperator<TradePaymentGMVBean> flatMapStream = keyedStream.flatMap(new FlatMapFunction<JSONObject, TradePaymentGMVBean>() {
            @Override
            public void flatMap(JSONObject value, Collector<TradePaymentGMVBean> out) throws Exception {
                HashSet<String> ss = new HashSet<>();
                String uid = value.getString("user_id");
                ss.add(uid);
                TradePaymentGMVBean.builder()
                        .final_amount(Long.parseLong(value.getString("final_amount")))
                        .pay_time(1L)
                        .sc(ss)
                        .build();
            }
        });

        /*//TODO 独立访客部分

        //TODO 3.1 过滤出首次登录页面的数据 不一定需要
        SingleOutputStreamOperator<JSONObject> jsonObjStream2 = tradePaymentStream2.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                String lastPageID = jsonObject.getJSONObject("page").getString("last_page_id");
                if (lastPageID == null) {
                    out.collect(jsonObject);
                }
            }
        });
        //TODO 3.2 按照设备id分组过滤掉相同id的数据
        KeyedStream<JSONObject, String> keyedStream2 = jsonObjStream2.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject value) throws Exception {
                return value.getJSONObject("common").getString("mid");
            }
        });
        SingleOutputStreamOperator<JSONObject> filterStream = keyedStream2.filter(new RichFilterFunction<JSONObject>() {
            ValueState<String> lastVisitDt = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> lastVisitDtDescriptor = new ValueStateDescriptor<>("lastVisitDt", String.class);
                lastVisitDtDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(org.apache.flink.api.common.time.Time.days(1L)).setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite).build());
                lastVisitDt = getRuntimeContext().getState(lastVisitDtDescriptor);
            }

            @Override
            public boolean filter(JSONObject value) throws Exception {
                String visitDt = DateFormatUtil.toDate(value.getLong("ts"));
                String lastDt = lastVisitDt.value();
                if (lastDt == null || !lastDt.equals(visitDt)) {
                    lastVisitDt.update(visitDt);
                }
                return false;
            }
        });

        SingleOutputStreamOperator<TradePaymentGMVBean> flatMapStream = keyedStream2.flatMap(new RichFlatMapFunction<JSONObject, TradePaymentGMVBean>() {
            ValueState<String> lastVisitDt = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> lastVisitDtDescriptor = new ValueStateDescriptor<>("lastVisitDt", String.class);
                lastVisitDtDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(org.apache.flink.api.common.time.Time.days(1L)).setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite).build());
                lastVisitDt = getRuntimeContext().getState(lastVisitDtDescriptor);
            }

            @Override
            public void flatMap(JSONObject value, Collector<TradePaymentGMVBean> out) throws Exception {
                String visitDt = DateFormatUtil.toDate(value.getLong("ts"));
                String lastDt = lastVisitDt.value();
                if (lastDt == null || !lastDt.equals(visitDt)) {
                    lastVisitDt.update(visitDt);
                    out.collect(TradePaymentGMVBean.builder()
                            .visit(1L)
                            .build()
                    );
                }
            }
        });

        //类型不相同<JSONObject,TradePaymentGMVBean>/<TradePaymentGMVBean>
        keyedStream.union(flatMapStream)*/



        //TODO 4 开窗统计
        SingleOutputStreamOperator<TradePaymentGMVBean> reduceStream = flatMapStream.windowAll(TumblingEventTimeWindows.of(Time.seconds(10L)))
                .reduce(new ReduceFunction<TradePaymentGMVBean>() {

                    @Override
                    public TradePaymentGMVBean reduce(TradePaymentGMVBean value1, TradePaymentGMVBean value2) throws Exception {
                        Set<String> sc1 = value1.getSc();
                        Set<String> sc2 = value2.getSc();
                        sc1.addAll(sc2);
                        value1.setHuman_count((long) sc1.size());
                        value1.setFinal_amount(value1.getFinal_amount() + value2.getFinal_amount());
                        value1.setPay_time(value1.getPay_time() + value2.getPay_time());
                        value1.setRate((value1.getHuman_count() + value2.getHuman_count()) / (value1.getPay_time() + value2.getPay_time()));
                        return value1;
                    }
                }, new AllWindowFunction<TradePaymentGMVBean, TradePaymentGMVBean, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow window, Iterable<TradePaymentGMVBean> values, Collector<TradePaymentGMVBean> out) throws Exception {
                        TradePaymentGMVBean tradePaymentGMVBean = values.iterator().next();
                        tradePaymentGMVBean.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                        tradePaymentGMVBean.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                        tradePaymentGMVBean.setTs(System.currentTimeMillis());
                        out.collect(tradePaymentGMVBean);
                    }
                });

        //TODO 5 写出到ClickHouse
        reduceStream.addSink(ClickHouseUtil.getClickHouseSinkFunc("insert into dws_trade_payment_GMV values(?,?,?,?,?,?,?)"));

        //TODO 6 执行任务
        env.execute(groupID);
    }
}
